package com.jml.demo001;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * kafka的原生API
 */
public class Producer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers","localhost:9092");
        props.put("key.serializer", StringSerializer.class);
        props.put("value.serializer", StringSerializer.class);
        //配置生产者的缓存容量,批量抓取
        props.put("buffer.memory", 524288);
        props.put("batch.size", 65536 );
        //不管batch满没满,都定时发送,单位毫秒,并发量大最好是是10或者20毫秒.
        props.put("linger.ms", 100);
        //该参数kafkamq服务器端限制接受的消息
        props.put("max.request.size", 10485760);
        //该参数设置定重试的次数、间隔时间
        props.put("retries", 10);
        props.put("retry.backoff.ms", 500);
        //服务器签收模式(0,1,-1)
        //0 不用等待broker恢复
        //1 只需要leader同步就行,当从节点还没同步leader就挂了,存在丢失消息的风险
        //-1 leader和follow节点都需要同步,存在耗时,效率不该
        //依据业务场景来决定
        props.put("ack", 1);
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        String msg = "jml666";
        ProducerRecord<String, String> record = new ProducerRecord<String, String>("mtntopic", "1", msg);
        producer.send(record);
        producer.close();
    }
}
